package org.zhadoop.streaming.util;

import java.math.BigDecimal;
import java.security.Timestamp;
import java.sql.Date;
import java.sql.Time;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.exception.ErrorCode;
import org.zhadoop.streaming.exception.StreamingException;
import org.zhadoop.streaming.util.datatype.BooleanParser;
import org.zhadoop.streaming.util.datatype.DataTypeParser;
import org.zhadoop.streaming.util.datatype.DateParser;
import org.zhadoop.streaming.util.datatype.DecimalParser;
import org.zhadoop.streaming.util.datatype.DoubleParser;
import org.zhadoop.streaming.util.datatype.FloatParser;
import org.zhadoop.streaming.util.datatype.IntParser;
import org.zhadoop.streaming.util.datatype.LongParser;
import org.zhadoop.streaming.util.datatype.StringParser;
import org.zhadoop.streaming.util.datatype.TimeParser;
import org.zhadoop.streaming.util.datatype.TimestampParser;

/**
 * Streaming 数据类型
 * @author zwd
 *
 */
public enum StreamingDataType {
	
	/**
	 * 字符串数据类型
	 */
	STRING("STRING",String.class,null){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new StringParser();
		}
		
	},
	
	/**
	 * int 数据类型
	 */
	INT("INT",Integer.class,int.class){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new IntParser();
		}
	},
	
	/**
	 * long 数据类型
	 */
	LONG("LONG",Long.class,long.class){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new LongParser();
		}
	},
	
	
	/**
	 * float 数据类型
	 */
	FlOAT("FLOAT",Float.class,float.class){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new FloatParser();
		}
	},
	
	/**
	 * double 类型
	 */
	DOUBLE("DOUBLE",Double.class,double.class) {
		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new DoubleParser();
		}
	},
	
	/**
	 * boolean 数据类型
	 */
	BOOLEAN("BOOLEAN",Boolean.class,boolean.class) {
		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new BooleanParser();
		}
	},
	
	/**
	 * 时间戳类型
	 */
	TIMESTAMP("TIMESTAMP",Timestamp.class,null){

		@Override
		public DataTypeParser createParser(StreamingConfig conf) throws StreamingException {
			return new TimestampParser(conf);
		}
	},
	
	/**
	 * 日期类型
	 */
	DATE("DATE",Date.class,null){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new DateParser(config);
		}
	},
	
	/**
	 * 时间类型
	 */
	TIME("TIME",Time.class,null){

		@Override
		public DataTypeParser createParser(StreamingConfig conf) throws StreamingException {
			return new TimeParser(conf);
		}
	},
	
	/**
	 * decimal
	 */
	DECIMAL("DECIMAL",BigDecimal.class,null){

		@Override
		public DataTypeParser createParser(StreamingConfig config) throws StreamingException {
			return new DecimalParser();
		}
		
	}
	;
	
	
	
	private static final Logger LOG = LoggerFactory.getLogger(StreamingDataType.class);
	
	/**
	 * 数据类型在CQL中的描述
	 * 所以该描述内容不允许随意修改
	 */
	private String desc;
	
	/**
	 * 包装类型
	 * 包装类型不能为空
	 */
	private Class<?> wrapperClass;
	
	/**
	 * 原始类型
	 */
	private Class<?> innerClass;
	
	StreamingDataType(String t,Class<?> wrapperClazz,Class<?> innerClazz) {
		this.desc = t;
		this.wrapperClass = wrapperClazz;
		this.innerClass = innerClazz;
	}

	public String getDesc() {
		return desc;
	}

	public Class<?> getWrapperClass() {
		return wrapperClass;
	}

	public Class<?> getInnerClass() {
		return innerClass;
	}
	
	/**
	 * 是否和当前枚举类型相同
	 * @param clazz
	 * @return
	 */
	private boolean isEqualDataType(Class< ? > clazz){
        return clazz.equals(wrapperClass) || clazz.equals(innerClass);
    }
	
	/**
	 * 通过Class 获取数据类型
	 * @param clazz
	 * @param conf
	 * @return
	 * @throws StreamingException
	 */
	public static DataTypeParser getDataTypeParser(Class<?> clazz,StreamingConfig conf)
	throws StreamingException{
		for(StreamingDataType dataType : StreamingDataType.values()){
			if(dataType.isEqualDataType(clazz)){
				return dataType.createParser(conf);
			}
		}
		StreamingException exception 
					= new StreamingException(ErrorCode.SEMANTICANALYZE_UNSUPPORTED_DATATYPE, clazz.getName()); 
		LOG.error("UnSupport data type.", exception);
        throw exception;
	}
	
	
	public static StreamingDataType getStreamingDataType(Class< ? > clazz) 
			throws StreamingException{
		 for (StreamingDataType dataType : StreamingDataType.values())
	        {
	            if (dataType.isEqualDataType(clazz))
	            {
	                return dataType;
	            }
	        }

	        StreamingException exception =
	         new StreamingException(ErrorCode.SEMANTICANALYZE_UNSUPPORTED_DATATYPE, clazz.getName());
	        LOG.error("UnSupport data type.", exception);
	        throw exception;
	}
	
	
	public abstract DataTypeParser createParser(StreamingConfig config) throws StreamingException;
	
	
}
